/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.tabletserver;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInput;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.SortedMap;
import java.util.SortedSet;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.nhncorp.neptune.changelog.ChangeLogInterruptedException;
import com.nhncorp.neptune.changelog.previous.ChangeLogStatus;
import com.nhncorp.neptune.client.Cell;
import com.nhncorp.neptune.client.CellFilter;
import com.nhncorp.neptune.client.Row;
import com.nhncorp.neptune.common.Latch;
import com.nhncorp.neptune.common.Constants;
import com.nhncorp.neptune.common.IdGenerator;
import com.nhncorp.neptune.common.LockServiceFactory;
import com.nhncorp.neptune.common.NeptuneLock;
import com.nhncorp.neptune.common.conf.NConfiguration;
import com.nhncorp.neptune.common.exception.ChangeLogException;
import com.nhncorp.neptune.fs.ChangeLogFileSystem;
import com.nhncorp.neptune.fs.ChangeLogFileSystemIF;
import com.nhncorp.neptune.fs.GPath;
import com.nhncorp.neptune.fs.NeptuneFileSystem;
import com.nhncorp.neptune.tablet.ColumnValue;
import com.nhncorp.neptune.tablet.TabletInfo;

public class MemorySSTable implements MemorySSTableIF {
  public static final Log LOG = LogFactory.getLog(MemorySSTable.class.getName());
  
  protected TabletInfo tabletInfo;
  
  protected int tabletSize;
  
  protected int tabletSizeBeforeCompaction;

  // ColumnName, ColumnCollection
  protected Map<String, ColumnCollection> columnCollections = Collections.synchronizedMap(new HashMap<String, ColumnCollection>(10));

  protected Map<String, ColumnCollection> compactingColumnCollections;
  
  protected final NeptuneLock lock = new NeptuneLock();
  
  protected boolean compacting = false;
  
  protected NeptuneFileSystem fs;
  
  protected NConfiguration conf;
  
  protected ChangeLogFileSystemIF changeLogFileSystem;

  //protected ChangeLogWriter logWriter;
  
  protected TabletServerIF tabletServer;

  private int numOfVersion;
  private Latch latch = new Latch();
  
  AtomicBoolean inBackupChangeLogStatus = new AtomicBoolean(false);
  AtomicInteger numOfCommitThreads = new AtomicInteger(0);
  
  public MemorySSTable() {
    
  }
  
  public void init(TabletServerIF tabletServer, NConfiguration conf, TabletInfo tabletInfo, int numOfVersion) throws IOException {
    this.tabletServer = tabletServer;
    this.tabletInfo = tabletInfo;
    this.numOfVersion = numOfVersion;
    this.conf = conf;
    this.fs = NeptuneFileSystem.get(conf);
    if(tabletServer != null) {
      try {
        this.changeLogFileSystem = ChangeLogFileSystem.getChangeLogFileSystem(conf, tabletServer,
                        LockServiceFactory.getLockService(conf, tabletServer.getHostName(), true));
      } catch(Exception e) {
        LOG.fatal("TabletServer is shut down due to fail initializing changelog filesystem", e);
        tabletServer.shutdown();
      }
    }

    //this.logWriter = new ChangeLogWriter(conf, changeLogFileSystem, tabletInfo);
    this.compactingColumnCollections = Collections.synchronizedMap(new HashMap<String, ColumnCollection>(10));
  }
  
  public void clearAllMemory() {
    columnCollections.clear();
    if(compactingColumnCollections != null) {
      compactingColumnCollections.clear(); 
    }
  }
  
  public Row.Key findMidRowKeyForSplit() throws IOException {
    lock.obtainReadLock("findMidRowKeyForSplit");
    try {
      int max = 0;
      ColumnCollection maxColumnCollection = null;
      for (Map.Entry<String, ColumnCollection> entry : columnCollections.entrySet()) {      
        ColumnCollection columnCollection = entry.getValue();
        if(columnCollection.getRowCount() > max) {
          max = columnCollection.getRowCount();
          maxColumnCollection = columnCollection;
        }
      }
      
      if(maxColumnCollection == null) {
        return null; 
      }
      
      Row.Key[] rowKeys = maxColumnCollection.columnCollectionMap.keySet().toArray(new Row.Key[]{});
      
      if(rowKeys.length == 2) {
        return rowKeys[0];
      } else {
        return rowKeys[rowKeys.length / 2];
      }
    } finally {
      lock.releaseReadLock("findMidRowKeyForSplit");
    }
  }
  
  public SortedSet<Row.Key> getAllRowKeys() {
    lock.obtainReadLock("getAllRowKeys");
    try {
      SortedSet<Row.Key> result = new TreeSet<Row.Key>(); 
      for (Map.Entry<String, ColumnCollection> entry : columnCollections.entrySet()) {      
        ColumnCollection columnCollection = entry.getValue();
        result.addAll(columnCollection.columnCollectionMap.keySet());
      }
      return result;
    } finally {
      lock.releaseReadLock("getAllRowKeys");
    }
  }
  
  public void print() {
    lock.obtainReadLock("print");
    try {
      for (Map.Entry<String, ColumnCollection> entry : columnCollections.entrySet()) {
        System.out.println("Column:" + entry.getKey());
        entry.getValue().print();
      }
      for (Map.Entry<String, ColumnCollection> entry : compactingColumnCollections.entrySet()) {
        System.out.println("Column:" + entry.getKey());
        entry.getValue().print();
      }
    } finally {
      lock.releaseReadLock("print");
    }
  }

  public int getTabletSize() {
    return tabletSize;
  }
  
  public int[] getDataCount() {
    lock.obtainReadLock("getDataCount");
    Set<String> columnNames;
    try {
      columnNames = columnCollections.keySet();
    } finally {
      lock.releaseReadLock("getDataCount");
    }  
    
    if(columnNames == null) {
      return new int[]{0};
    }
    
    int result[] = new int[columnNames.size()];
    int index = 0;
    for(String column: columnNames) {
      ColumnCollection columnCollection = columnCollections.get(column);
      if(columnCollection != null) {
        result[index] = columnCollection.columnCollectionMap.size();
      }
      index++;
    }
    return result;
      
  }
  
  public ChangeLogFileSystemIF getChangeLogFileSystem() {
    return changeLogFileSystem;
  }
  
  public Map<String, ColumnCollection> getColumnCollections() {
    return columnCollections;
  }
  
  public Map<String, ColumnCollection> getCompactingColumnCollections() {
    return compactingColumnCollections;
  }

  public ColumnValue[] getAllMemoryValues(String columnName) {
    lock.obtainReadLock("getAllMemoryValues");
    try {
      ColumnCollection columnCollecition = columnCollections.get(columnName);
      if(columnCollecition == null) {
        return null;
      }
      
      List<ColumnValue> columnValues = new ArrayList<ColumnValue>();
      
      for(SortedMap<Cell.Key, ValueCollection> eachValue: columnCollecition.columnCollectionMap.values()) {
        for(ValueCollection eachValueCollection: eachValue.values()) {
          ColumnValue columnValue = eachValueCollection.get();
          if(columnValue != null) {
            columnValues.add(columnValue);
          }
        }
      }
      
      if(columnValues.size() == 0) {
        return null;
      } else {
        return columnValues.toArray(new ColumnValue[columnValues.size()]);
      }
    } finally {
      lock.releaseReadLock("getAllMemoryValues");
    }
  }
  
  public void commit(TxId txId, ChangeLog[] changeLogList, boolean saveLog) throws IOException {
    if(saveLog) {
      String tabletName = tabletInfo.getTabletName();
      String txIdStr = txId.getTxId();

      //ChangeLog 파일 오픈
      //long beforeOpen = System.nanoTime();
      try {
        latch.enter();
        // by sangchul
        // 발생할 수 있는 Exception
        // ChangeLogInterruptedException : IO 연산 중 쓰레드가 interrupt 걸렸을 때
        // ChangeLogException : Pipe 여는 도중 작업을 취소하고 Minor compaction이 수행되어야 할 때
        // IOException : 연산 도중 에러가 발생했을 때.
        changeLogFileSystem.open(tabletName, true);
      } catch (InterruptedException e) {
        throw new ChangeLogInterruptedException();
      } catch (ChangeLogInterruptedException e) {
        latch.exit();
        throw e;
      } catch (IOException e) {
        LOG.error("changeLogFileSystem.open() error : " + e);
        latch.exit();
        throw e;
      }
      
      //ChangeLog에 추가
      // by sangchul
      // ChangeLogInterruptedException : IO 연산 중 쓰레드가 interrupt 걸렸을 때
      // IOException : 연산 도중 에러가 발생했을 때.
      try {
        int index = 1;
        for(ChangeLog changeLog: changeLogList) {
          changeLogFileSystem.addChangeLog(tabletName, txIdStr, index++, changeLog);
        }
        changeLogFileSystem.finishAdding(tabletName, txIdStr);

      } catch (ChangeLogInterruptedException e) {
        throw e;
      } catch (IOException e) {
        LOG.error("Change Log Error in tablet [" + tabletInfo.getTabletName() + "], txid [" + txId + "], exception : " + e);
        throw new ChangeLogException(e);
      } finally {
        latch.exit();
      }
      
      //메모리에 commit된 값 추가
      long lockStartTime = System.currentTimeMillis();
      lock.obtainWriteLock("commit");
      long lockLapTime = System.currentTimeMillis() - lockStartTime;
      if(lockLapTime > 100) {
        LOG.warn("Too long lock time:" + lockLapTime + "," + tabletInfo.getTabletName());
        lock.printLockOwner();
      }
      try {
        for(ChangeLog changeLog: changeLogList) {
          addValue(changeLog);
        }
      } catch (Exception e) {
        //FIXME 장애 발생시 change log에 대한 처리는?? 
        e.printStackTrace(System.out);
        throw new IOException(e.getMessage());
      } finally {
        lock.releaseWriteLock("commit");
      }
    } else {
      lock.obtainWriteLock("commit");
      try {
        for(ChangeLog changeLog: changeLogList) {
          addValue(changeLog);
        }
      } finally {
        lock.releaseWriteLock("commit");
      }
    }
  }

  public void clearChangeLogOf(String tabletName) {
    LOG.info("remove all the change logs of tablet [" + tabletName + "]");
    try {
      changeLogFileSystem.delete(tabletName);
    } catch (IOException e) {
      LOG.warn("Exception in deleting change logs of tablet[" + tabletName + "] due to " + e);
    }    
  }
  
  public void endChangeLogMinorCompaction() throws IOException {
    changeLogFileSystem.endMinorCompaction(tabletInfo.getTabletName());
  }
  
  /**
   * false가 return되면 로딩 후 바로 minorcompaction을 수행해야 한다.
   */
  public boolean loadFromChangeLog() throws IOException {
    if(changeLogFileSystem == null) {
      return true;
    }
    String tabletName = tabletInfo.getTabletName();

    if(!changeLogFileSystem.exists(tabletName)) {
      return true;
    }

    ChangeLogStatus changeLogStatus = changeLogFileSystem.verifyChangelog(tabletName);
    String changeLogServer = null;
    if(changeLogStatus != null) {
      //changelog 서버들 사이의 버전이 맞지 않는 경우 
      //반환된 changelog server로 부터 로그 데이터를 읽어 메모리를 구성한 다음
      //minor compaction을 수행시키도록 한다.
      changeLogServer = changeLogStatus.getHostName();
    } else {
      throw new IOException("Error while Verify Changelog for :" + tabletName);
    }
    
    lock.obtainWriteLock("loadFromChangeLog");
    try {
      changeLogFileSystem.open(changeLogServer, tabletName, false);
      ChangeLog changeLog = new ChangeLog();
      int count = 0;
      while((changeLog = changeLogFileSystem.nextChangeLog(tabletName)) != null) {
        //System.out.println("changeLog:" + changeLog);
        addValue(changeLog);
        count++;
      }
      LOG.info(count + " change logs are loaded");
    } catch (IOException e) {
      //FIXME Changelog 파일에 문제가 있는 경우 전체 시스템에 경고를 주고, 매뉴얼하게 fsck하는 기능이 있어야 한다.
      LOG.fatal("Check change log file:" + tabletName);
      e.printStackTrace(System.out);
      throw e;
    } finally {
      lock.releaseWriteLock("loadFromChangeLog");
      //changeLogFileSystem.close(tabletName, false);
    }
    
    if(changeLogServer != null && changeLogStatus.isNeedCompaction()) {
      return false;
    } else {
      return true;
    }  
  }
  
  private void addValue(ChangeLog changeLog) throws IOException {
    if(changeLog.getOperation() == Constants.LOG_OP_MODIFY_META) {
      addMetaValue(changeLog);
    } else {
      String columnName = changeLog.getColumnName();
      ColumnCollection columnCollection = columnCollections.get(columnName);
      //if null, first insert in column
      if (columnCollection == null) {
        columnCollection = new ColumnCollection(conf);
        columnCollections.put(columnName, columnCollection);
      }
      boolean deleted = (changeLog.getOperation() == Constants.LOG_OP_DELETE_COLUMN_VALUE);
      
      ColumnValue columnValue = new ColumnValue(changeLog.getRowKey(), 
          changeLog.getCellKey(), 
          changeLog.getValue(), 
          deleted, 
          changeLog.getTimestamp());
      
      tabletSize += columnCollection.addValue(changeLog.getRowKey(), columnValue, numOfVersion);
    }
    //tabletSize += changeLog.getByteSize();
  }

  /**
   * change log가 splited된 tablet에 대한 meta 데이터 변경 작업인 경우
   * @param changeLog
   */
  private void addMetaValue(ChangeLog changeLog) throws IOException {
    DataInput in = new DataInputStream(new ByteArrayInputStream(changeLog.getValue()));
    TabletInfo targetTablet = new TabletInfo();
    targetTablet.readFields(in);
    TabletInfo[] splitedTabletInfos = new TabletInfo[2];
    for(int i = 0; i < 2; i++) {
      splitedTabletInfos[i] = new TabletInfo();
      splitedTabletInfos[i].readFields(in);
//      System.out.println("addMetaValue>>>>" + splitedTabletInfos[i]);
    }
    long timestamp = changeLog.getTimestamp();

    Row.Key rowKey = changeLog.getRowKey();

    //delete split target tablet
    ByteArrayOutputStream bout = new ByteArrayOutputStream();
    targetTablet.write(new DataOutputStream(bout));
    
    ColumnValue columnValue = new ColumnValue(rowKey, new Cell.Key(targetTablet.getTabletName()), bout.toByteArray(), true, timestamp);
    
    ColumnCollection columnCollection = columnCollections.get(Constants.META_COLUMN_NAME_TABLETINFO);
    //if null, first insert in column
    if (columnCollection == null) {
      columnCollection = new ColumnCollection(conf);
      columnCollections.put(Constants.META_COLUMN_NAME_TABLETINFO, columnCollection);
    }
    columnCollection.addValue(rowKey, columnValue, numOfVersion);

    //insert splited tablet
    for(int i = 0; i < 2; i++) {
      timestamp++;
      
      rowKey = Tablet.generateMetaRowKey(splitedTabletInfos[i].getTableName(), splitedTabletInfos[i].getEndRowKey());
      
      bout = new ByteArrayOutputStream();
      splitedTabletInfos[i].write(new DataOutputStream(bout));

      columnValue = new ColumnValue(rowKey, new Cell.Key(splitedTabletInfos[i].getTabletName()), bout.toByteArray(), false, timestamp);
      
      columnCollection = columnCollections.get(Constants.META_COLUMN_NAME_TABLETINFO);
      //if null, first insert in column
      if (columnCollection == null) {
        columnCollection = new ColumnCollection(conf);
        columnCollections.put(Constants.META_COLUMN_NAME_TABLETINFO, columnCollection);
      }
      tabletSize += columnCollection.addValue(rowKey, columnValue, numOfVersion);
    }  
  }
  
  public ColumnValue search(Row.Key rowKey, String columnName, Cell.Key columnKey) {
    lock.obtainReadLock("search");
    try {
      ColumnCollection columnCollection = columnCollections.get(columnName);
      if(columnCollection != null) {
        ColumnValue columnValue = columnCollection.get(rowKey, columnKey);
        if(columnValue != null)  {
          return columnValue.copyColumnValue();
        }
      }
      
      columnCollection = compactingColumnCollections.get(columnName);
      if(columnCollection != null) {
        ColumnValue columnValue = columnCollection.get(rowKey, columnKey);
        if(columnValue != null)  return columnValue.copyColumnValue();
      }
      return null;
    } finally {
      lock.releaseReadLock("search");
    }
  }
  
  protected List<ColumnValue> searchAllVersion(Row.Key rowKey, String columnName, Cell.Key columnKey) throws IOException {
    lock.obtainReadLock("searchAllVersion");
    try {    
      //columnCollections에 있는 정보가 최신 정보
      ColumnCollection columnCollection = columnCollections.get(columnName);
      List<ColumnValue> result = new ArrayList<ColumnValue>();
      if(columnCollection != null) {
        ValueCollection vc = columnCollection.getValueCollection(rowKey, columnKey);
        if(vc != null) {
          for(ColumnValue eachColumnValue: vc.getColumnRecords()) {
            result.add(eachColumnValue.copyColumnValue());
          }
        }
      }
      
      columnCollection = compactingColumnCollections.get(columnName);
      if(columnCollection != null) {
        ValueCollection vc = columnCollection.getValueCollection(rowKey, columnKey);
        if(vc != null) {
          for(ColumnValue eachColumnValue: vc.getColumnRecords()) {
            result.add(eachColumnValue.copyColumnValue());
          }
        }
      }
      
      if(result.size() == 0) {
        return null;
      } else {
        return result;
      }
    } finally {
      lock.releaseReadLock("searchAllVersion");
    }
  }
  
  
  /**
   * 결과는 순서로 정렬되어 있지 않다. Tablet의 RecordSearcher에서 정렬한다.
   */
  public Collection<ColumnValue> search(Row.Key rowKey, String columnName) throws IOException {
    lock.obtainReadLock("search");
    try {  
      List<ColumnValue> result = new ArrayList<ColumnValue>(10);
      
      //compactingColumnCollections, columnCollections 두군데 동일한 Cell.Key값을 가지는 레코드가 있는 경우
      //columnCollections에 있는 값이 최신 값이다.
      //따라서 columnCollections에 대한 처리를 뒤에 한다.
      ColumnCollection columnCollection = compactingColumnCollections.get(columnName);
      
      if(columnCollection != null) {
        SortedMap<Cell.Key, ValueCollection> rowDatas = columnCollection.columnCollectionMap.get(rowKey);
        if(rowDatas != null) {
          for(ValueCollection vc: rowDatas.values()) {
            //RPC 속도에 최적화하기 위해 RPC 전송전 columnValue의 값이 변경될 수 있는데
            //동일한 레퍼런스를 사용할 경우 MemorySSTable의 값도 변경되기 때문에 복사해서 사용
            for(ColumnValue eachColumnValue: vc.getColumnRecords()) {
              result.add(eachColumnValue.copyColumnValue());
            }
          }
        }
      }
    
      columnCollection = columnCollections.get(columnName);
      
//      String inputKey = new String(rowKey.getBytes(), "EUC-KR");
//      if(columnCollection != null && columnCollection.columnCollectionMap != null) {
//        for(Row.Key key: columnCollection.columnCollectionMap.keySet()) {
//          LOG.info("Key0000:" + new String(key.getBytes(), "EUC-KR") + "," + inputKey + "," + key.equals(rowKey) + "," + rowKey.getLength() + "," + key.getLength());
//        }
//      }
      if(columnCollection != null) {
        SortedMap<Cell.Key, ValueCollection> rowDatas = columnCollection.columnCollectionMap.get(rowKey);
        if(rowDatas != null) {
          for(ValueCollection vc: rowDatas.values()) {
            //RPC 속도에 최적화하기 위해 RPC 전송전 columnValue의 값이 변경될 수 있는데
            //동일한 레퍼런스를 사용할 경우 MemorySSTable의 값도 변경되기 때문에 복사해서 사용
            for(ColumnValue eachColumnValue: vc.getColumnRecords()) {
              ColumnValue copyedValue = eachColumnValue.copyColumnValue();
              result.add(copyedValue);
//              System.out.println("aaaaaa:" + copyedValue);
            }
          }
        }
      }
      
      return result;
    } finally {
      lock.releaseReadLock("search");
    }
  }

  public boolean isDeleted(Row.Key rowKey, String columnName) throws IOException {
    Collection<ColumnValue> memoryValues = search(rowKey, columnName);
    if(memoryValues.size() == 0) {
      return false;
    }
    ValueCollection valueCollection = new ValueCollection();
    for(ColumnValue eachColumnValue: memoryValues) {
      valueCollection.add(eachColumnValue, 0);
    }    
    if(valueCollection.getValueSize() == 0) {
      return false;
    }
    return valueCollection.get() == null;
  }
  
  public ColumnValue findClosest(Row.Key rowKey, String columnName) {
    lock.obtainReadLock("findClosest");
    try {  
      
      ColumnCollection columnCollection = columnCollections.get(columnName);
      ColumnValue result = null;
      if(columnCollection != null) {
        result = columnCollection.findNearestValue(rowKey);
      } 
  
      if(compactingColumnCollections != null) {
        columnCollection = compactingColumnCollections.get(columnName);
        if(columnCollection != null) {
          ColumnValue otherResult = columnCollection.findNearestValue(rowKey);
          if(otherResult != null) {
            if(result == null || otherResult.getRowKey().compareTo(result.getRowKey()) < 0) {
              result = otherResult;
            }
          }
        }
      } 
      //LOG.debug("findNearest: " + rowKey + ", columnName:" + columnName + ", result=" + resultRow.Key);
      return result;
    } finally {
      lock.releaseReadLock("findClosest");
    }
  }
  
  public boolean hasValue(Row.Key rowKey, String columnName) throws IOException {
    Collection<ColumnValue> memoryValues = search(rowKey, columnName);
    if(memoryValues.size() == 0) {
      return false;
    }
    ValueCollection valueCollection = new ValueCollection();
    for(ColumnValue eachColumnValue: memoryValues) {
      valueCollection.add(eachColumnValue, 3);
    }    
    return valueCollection.get() != null;
  }
  
  /**
   * 메모리에 있는 내용을 임시버퍼로 저장시킨 다음, 비어 있는 새로운 메모리를 구성한다.
   * 임시버퍼에 저장된 내용을 파일로 저장한다.
   */
  public void initMemoryForCompaction() throws IOException {
    lock.obtainWriteLock("initMemoryForCompaction");
    try {
      try {
        latch.block();
        changeLogFileSystem.startMinorCompaction(tabletInfo.getTabletName());
      } catch(InterruptedException e) {
        LOG.info("MinorCompaction is interrupted in initMemoryForCompaction");
        throw new IOException(e);
      } finally {
        latch.unblock();
      }
      
      compactingColumnCollections = columnCollections;
      columnCollections = new HashMap<String, ColumnCollection>(10);
      tabletSizeBeforeCompaction = tabletSize;
      tabletSize = 0;
    } finally {
      compacting = true;
      lock.releaseWriteLock("initMemoryForCompaction");
    }
  }
  
  void setBackupChangeLogStatus(boolean flag) {
    synchronized(inBackupChangeLogStatus) {
      inBackupChangeLogStatus.set(flag);
      inBackupChangeLogStatus.notifyAll();
    }
  }

  /**
   * Compaction 수행 중 오류가 발생했을 때 메모리를 원상복구 시킨다.
   *
   */
  public void cancelCompaction(String fileId) {
    lock.obtainWriteLock("cancelCompaction");
    try {
      //tempCollections에 저장하는 순서 중요
      //compactingColumnCollections에 저장된 데이터가 더 오래된 버전
      Map<String, ColumnCollection> tempCollections = compactingColumnCollections;

      for (Map.Entry<String, ColumnCollection> entry : columnCollections.entrySet()) {
        String columnName = entry.getKey();
        ColumnCollection entryColumnCollection = entry.getValue(); 
        ColumnCollection tempColumnCollection = tempCollections.get(columnName); 
        if(tempColumnCollection == null) {
          tempCollections.put(columnName, entryColumnCollection);
        } 
      }
      
      columnCollections = tempCollections;
      compactingColumnCollections.clear();
      
      tabletSize = tabletSize + tabletSizeBeforeCompaction;
      tabletSizeBeforeCompaction = 0;
      
      if(fileId != null) {
        //temp 파일 삭제
        for (Map.Entry<String, ColumnCollection> entry : columnCollections.entrySet()) {
          String columnName = entry.getKey();
          GPath tempMapFilePath = new GPath(Tablet.getTabletMinorCompactionTempPath(conf, tabletInfo), columnName + "/" + fileId);
          try {
            if(!fs.delete(tempMapFilePath, true)) {
              LOG.error("Can't delete temp minor compaction data while cancel comapction:" + tempMapFilePath);
            }
          } catch (IOException e) {
            LOG.warn("Can't delete temp minor compaction data while cancel comapction:" + 
                tempMapFilePath + "," + e.getMessage(), e);
          }
        }
      }
    } finally {
      lock.releaseWriteLock("cancelCompaction");
    }
  }

  /**
   * 임시버퍼(compactingColumnCollections)에 저장된 데이터를 파일로 저장한다.
   * @param tabletInfo
   * @return
   * @throws IOException
   */
  public Map<String, TabletMapFile> saveToDisk(TabletInfo tabletInfo, String fileId) throws IOException {
    //compactingColumnCollections에 대한 변경은 Tablet에서 동시에 하나의 Thread만 처리하도록 되어 있기 때문에 동기화 필요 없음
    Map<String, TabletMapFile> result = new HashMap<String, TabletMapFile>(10);
    
    //Compaction처리 후 생성되는 FileId는 모든 컬럼이 동일하다.
    //이유는 ChangeLog 처리도중 발생하는 오류에 대해 recover하기 위해 changeLog를 임시로 fileId 디렉토리에 이동시킨다.
    for (Map.Entry<String, ColumnCollection> entry : compactingColumnCollections.entrySet()) {
      String columnName = entry.getKey();
      TabletMapFile mapFile = entry.getValue().saveToDisk(tabletInfo, columnName, fileId, numOfVersion);
      if(mapFile != null) {
        result.put(columnName, mapFile);
      }
    }
    return result;
  }
  
  public void obtainWriteLock(String owner) {
    lock.obtainWriteLock(owner);
  }
  
  public void releaseWriteLock(String owner) {
    lock.releaseWriteLock(owner);
  }
  
  public void clearCompactingColumnCollections() {
    this.compactingColumnCollections.clear();    
    this.compactingColumnCollections = null;
    this.compactingColumnCollections = Collections.synchronizedMap(new HashMap<String, ColumnCollection>(10));
    this.tabletSizeBeforeCompaction = 0;
  }
  
  public Scanner getScanner(String columnName, Row.Key startRowKey, 
      Row.Key endRowKey, CellFilter columnFilter) throws IOException {
    return new MemoryScanner(columnName, startRowKey, endRowKey, columnFilter);
  }
  
  public Searchable getSearcher(Row.Key rowKey, String columnName, Cell.Key columnKey) throws IOException {
    return new MemorySearcher(rowKey, columnName, columnKey); 
  }
  
  class MemorySearcher implements Searchable {
    Row.Key rowKey;
    String columnName;
    Cell.Key columnKey;
    Iterator<ColumnValue> columnValueIt;
    Collection<ColumnValue> columnValues;
    
    public MemorySearcher(Row.Key rowKey, String columnName, Cell.Key columnKey) throws IOException {
      this.rowKey = rowKey;
      this.columnName = columnName;
      this.columnKey = columnKey;
      
      if(columnKey == null) {
        this.columnValues = search(rowKey, columnName);
      } else {
        this.columnValues = searchAllVersion(rowKey, columnName, columnKey);
      }
      if(columnValues != null) {
        columnValueIt = columnValues.iterator();
      }
    }
    
//    public ColumnValue[] nextAll() throws IOException {
//      if(columnValues == null || columnValues.size() == 0) {
//        return null;
//      }
//      
//      return columnValues.toArray(new ColumnValue[]{});
//    }
    
    public ColumnValue next() throws IOException {
      if(columnValueIt == null || !columnValueIt.hasNext()) {
        columnValueIt = null;
        return null;
      }

      ColumnValue result = columnValueIt.next();
      return result;
    }
    
    public void close() throws IOException {
      columnValueIt = null;
      columnValues = null;
    }
  }
  
  /**
   * Scanner는 major, minor compaction 중인 경우에는 생성하지 못한다.
   * 또한 Scanner가 오픈되어 있으면 minor, major compaction은 수행하지 않는다.
   * MemoryScanner 객체는 생성 시 scan 범위에 해당하는 데이터만큼 메모리에 복사한다.(메모리 낭비가 심하다-> 튜닝필요)
   * 이렇게 하는 이유는 Scanner가 open된 이후 계속해서 트렌젝션이 들어오는 경우 트렌젝션과 분리시키기 위해서이다.
   * @author babokim
   *
   */
  class MemoryScanner implements Scanner {
    //FIXME Scan의 범위가 startRowKey, endRowKey내에서만 되도록.
    private Row.Key startRowKey;
    private Row.Key endRowKey;
    
    private Cell.Key startCellKey;
    
    private String columnName;

    private SortedMap<Row.Key, List<ColumnValue>> columnValues;
    
    private Iterator<Row.Key> rowKeyIt;
    
    private Iterator<ColumnValue> columnValueIt;
    
    private CellFilter cellFilter;
    
    public MemoryScanner(String columnName, Row.Key startRowKey, 
        Row.Key endRowKey, CellFilter cellFilter) throws IOException {
      this(columnName, startRowKey, Cell.Key.EMPTY_KEY, endRowKey, cellFilter);
    }

    public MemoryScanner(String columnName, Row.Key startRowKey, Cell.Key cellKey, 
                                  Row.Key endRowKey, CellFilter cellFilter) throws IOException {
      this.cellFilter = cellFilter;
      this.startRowKey = startRowKey;
      this.endRowKey = endRowKey;
      this.columnName = columnName;
      this.startCellKey = cellKey;
      this.columnValues = Collections.synchronizedSortedMap(new TreeMap<Row.Key, List<ColumnValue>>());
      
      lock.obtainReadLock("MemoryScanner");
      try {
        if(!columnCollections.containsKey(columnName)) {
          return;
        }

        copyValues();
        
        if(columnValues != null && columnValues.size() > 0) {
          rowKeyIt = columnValues.keySet().iterator();
          while(rowKeyIt.hasNext()) {
            Row.Key scanRowKey = rowKeyIt.next();
            List<ColumnValue> list = columnValues.get(scanRowKey);
            if(list != null) {
              columnValueIt = list.iterator();
              if(columnValueIt.hasNext()) {
                break;
              } 
            }
          } 
        }
      } catch (Exception e) {
        LOG.error("Error init Memory Scanner:" + e.getMessage(), e);
        IOException err = new IOException(e.getMessage());
        err.initCause(e);
        throw err;
      } finally {
        lock.releaseReadLock("MemoryScanner");
      }
    }

    private void copyValues() {
      SortedMap<Row.Key, SortedMap<Cell.Key, ValueCollection>> targetColumnValues = 
                    columnCollections.get(columnName).columnCollectionMap;
      
      for(Map.Entry<Row.Key, SortedMap<Cell.Key, ValueCollection>> entry: targetColumnValues.tailMap(startRowKey).entrySet()) {
        Row.Key rowKey = entry.getKey();
        if(rowKey.compareTo(endRowKey) > 0) {
          break;
        }
        SortedMap<Cell.Key, ValueCollection> entryValue = entry.getValue();
        
        List<ColumnValue> rowColumnValues = Collections.synchronizedList(new ArrayList<ColumnValue>());
        for(Map.Entry<Cell.Key, ValueCollection> entryValueEntry: entryValue.tailMap(startCellKey).entrySet()) {
          if(entryValueEntry.getValue().columnValues != null) {
            rowColumnValues.addAll(entryValueEntry.getValue().columnValues);
          }
        }
        columnValues.put(rowKey, rowColumnValues);
      }
    }
    
    public ColumnValue next() throws IOException {
      if(columnValues == null || columnValueIt == null) {
        return null;
      }

      if(columnValueIt.hasNext()) {
        return columnValueIt.next();
      } else {
        //현재의 rowKey에 더이상 column 값이 없는 경우, 다음 rowKey 데이터 조회
        while(rowKeyIt.hasNext()) {
          Row.Key scanRowKey = rowKeyIt.next();
          List<ColumnValue> list = columnValues.get(scanRowKey);
          if(list != null) {
            columnValueIt = list.iterator();
            if(columnValueIt.hasNext()) {
              return columnValueIt.next();
            } 
          }
        } 
        return null;
      }
    }

    public void close() throws IOException {
      if(columnValues != null) {
        columnValues.clear();
        columnValues = null;
      }
    }
  }

  public Map<String, TabletMapFile[]> splitAndSaveCanCommit(Row.Key midRowKey, TabletInfo[] splitedTabletInfos) throws IOException {
    initMemoryForCompaction();
    try {
      return splitAndSave(midRowKey, splitedTabletInfos, compactingColumnCollections, false);
    } catch (Exception e) {
      cancelCompaction(null);
      if(e instanceof IOException) {
        throw (IOException)e;
      } else {
        IOException err = new IOException(e.getMessage());
        err.initCause(e);
        throw err;
      }
    }
  }
  
  public Map<String, TabletMapFile[]> splitAndSave(Row.Key midRowKey, TabletInfo[] splitedTabletInfos) throws IOException {
    return splitAndSave(midRowKey, splitedTabletInfos, columnCollections, true);
  }
  
  private Map<String, TabletMapFile[]> splitAndSave(Row.Key midRowKey, 
      TabletInfo[] splitedTabletInfos, Map<String, ColumnCollection> columnCollections, boolean doLock) throws IOException {
    long startTime = System.currentTimeMillis();
    Map<String, TabletMapFile[]> result = new HashMap<String, TabletMapFile[]>(10);
    
    String fileId = IdGenerator.getId();

    if(doLock) {
      lock.obtainWriteLock("splitAndSave");
    }
    try {
      for (Map.Entry<String, ColumnCollection> entry : columnCollections.entrySet()) {
        String columnName = entry.getKey();
        ColumnCollection columnCollection = entry.getValue();
        if(columnCollection.getRowCount() == 0) {
          continue;
        }
        TabletMapFile[] mapFiles = columnCollection.splitAndSaveToDisk(tabletInfo, midRowKey, splitedTabletInfos, columnName, fileId, numOfVersion);
        result.put(columnName, mapFiles);
      }
      return result;
    } finally {
      if(doLock) {
        lock.releaseWriteLock("splitAndSave");
      }
      LOG.debug("splitAndSave:" + tabletInfo.getTabletName() + ", time=" + (System.currentTimeMillis() - startTime) + ",doLock=" + doLock);
    }
  }

  /**
   * @param compacting the compacting to set
   */
  public void setCompacting(boolean compacting) {
    this.compacting = compacting;
  }

  public boolean isEmpty() {
    lock.obtainReadLock("isEmpty");
    try {
      return columnCollections.size() == 0 && compactingColumnCollections.size() == 0;
    } finally {
      lock.releaseReadLock("isEmpty");
    }      
  }
  
  public String getTestHandlerKey() {
    return tabletServer.getHostName();
  }
  
  public TabletServerIF getTabletServer() {
    return tabletServer;
  }
  
  public static void main(String[] args) throws IOException {
    NConfiguration conf = new NConfiguration();
    ChangeLogFileSystemIF changeLogFileSystem = ChangeLogFileSystem.getChangeLogFileSystem(conf, null, null);
    
    String tabletName = args[0];
    
    if(!changeLogFileSystem.exists(tabletName)) {
      System.out.println("No change log");
    }

    ChangeLogStatus changeLogStatus = changeLogFileSystem.verifyChangelog(tabletName);
    String changeLogServer = null;
    if(changeLogStatus != null) {
      changeLogServer = changeLogStatus.getHostName();
    } else {
      throw new IOException("Error while Verify Changelog for :" + tabletName);
    }
    
    try {
      changeLogFileSystem.open(changeLogServer, tabletName, false);
      ChangeLog changeLog = new ChangeLog();
      int count = 0;
      while((changeLog = changeLogFileSystem.nextChangeLog(tabletName)) != null) {
        System.out.println("changeLog:" + changeLog);
        count++;
      }
      LOG.info(count + " change logs are loaded");
    } catch (IOException e) {
      e.printStackTrace();
    }
  }
}
